package net.bw.realtime.jtp.dws.log.function;

import net.bw.realtime.jtp.common.utils.DateTimeUtil;
import net.bw.realtime.jtp.dws.log.bean.PageViewBean;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/*
 * @author liuyawei
 * @date 2025/5/20
 */
public class PageViewReportWindowFunction implements WindowFunction<PageViewBean, String, String, TimeWindow> {
    @Override
    public void apply(String s, TimeWindow window, Iterable<PageViewBean> input, Collector<String> out) throws Exception {

        // 1.获取窗口开始和结束时间
        String windowStartTime = DateTimeUtil.convertLongToString(window.getStart(), DateTimeUtil.DATE_TIME_FORMAT);
        String windowEndTime = DateTimeUtil.convertLongToString(window.getEnd(), DateTimeUtil.DATE_TIME_FORMAT);

        // 2.计算
        PageViewBean pageViewBean = input.iterator().next();
        Long pvCount = pageViewBean.getPvCount();
        Long pvDuringTime = pageViewBean.getPvDuringTime();
        Long uvCount = pageViewBean.getUvCount();
        Long sessionCount = pageViewBean.getSessionCount();

        // 3.拼接数据
        String s1 = windowStartTime + "," + windowEndTime + "," + s + ","
                + pvCount + "," + pvDuringTime + "," + uvCount + "," + sessionCount + ","
                + System.currentTimeMillis();

        // 4.输出数据
        out.collect(s1);

    }
}
